package com.llx.dataset

import java.lang

import org.apache.flink.api.common.functions.{GroupCombineFunction, GroupReduceFunction}
import org.apache.flink.util.Collector

import scala.collection.mutable

class Tuple3GroupReduceWithCombine  extends GroupReduceFunction[( String , Int), (String, Int)] with GroupCombineFunction[(String, Int), (String, Int)] {
  /**
    * java集合转Scala需要 class 内部
    * import scala.collection.JavaConverters._
    */
  import scala.collection.JavaConverters._

  override def reduce(iterable: lang.Iterable[(String, Int)], collector: Collector[(String, Int)]): Unit ={
    for(in <- iterable.asScala){
      collector.collect((in._1,in._2))
    }
  }

  /**
    * 使用之前的group操作，比如：reduceGroup或者GroupReduceFuncation；这种操作很容易造成内存溢出；因为要一次性把所有的数据一步转化到位，所以需要足够的内存支撑，如果内存不够的情况下，那么需要使用combineGroup；
    * combineGroup在分组数据集上应用GroupCombineFunction。
    * GroupCombineFunction类似于GroupReduceFunction，但不执行完整的数据交换。
    * 【注意】：使用combineGroup可能得不到完整的结果而是部分的结果
    *
    * @param iterable
    * @param collector
    */
  override def combine(iterable: lang.Iterable[(String, Int)], collector: Collector[(String, Int)]): Unit = {
    val map = new mutable.HashMap[String,Int]
    var num = 0
    var s = ""
    for(in <- iterable.asScala){
      num += in._2
      s = in._1
    }
    collector.collect((s , num))
  }


}
